package com.atguigu.sink;

import com.atguigu.func.ClickSource;
import com.atguigu.pojo.Event;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @Author Mr.Zheng
 * @Date 2023/6/15 11:01
 *
 *  JdbcSink
 *      1. 常规写入
 *      2. exactilyOnceSink() 支持分布式写入，支持分布式事务， 支持EOS
 *
 * JDBC步骤：
 *      加载驱动
 *      获取链接
 *      编写SQL
 *      预编译SQL
 *      设置参数
 *      执行SQL
 *      处理结果
 *      关闭链接
 *
 * 写入方式：
 *      1. 直接写入  insert into clicks （XXX） values （XXX）
 *      2. 幂等写入
 *          replace into clicks(XXX) value (XXX)
 *          insert into clicks(XXXX) values ( XXX) on Duplicate key update
 *
 */
public class Flink04_JdbcSink {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Event> ds = env.addSource(new ClickSource());
        ds.print("input");

        //将流中的数据写入到Mysql
        SinkFunction<Event> jdbcSink = JdbcSink.<Event>sink(
                //"insert into clicks1 (user ,url ,ts ) values(?,?,?)",
                "replace into clicks(user, url, ts ) values(?,?,?)" ,
                new JdbcStatementBuilder<Event>() {
                    @Override
                    public void accept(PreparedStatement preparedStatement, Event event) throws SQLException {
                        preparedStatement.setString(1, event.getUser());
                        preparedStatement.setString(2, event.getUrl());
                        preparedStatement.setLong(3, event.getTs());
                    }
                },

                JdbcExecutionOptions.builder()
                        .withBatchSize(5)  // 批写大小
                        .withBatchIntervalMs(10000)  // 批次间隔时间， 超过指定时间也要写入
                        .withMaxRetries(3)  //重试次数
                        .build(),

                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName("com.mysql.cj.jdbc.Driver")
                        .withUrl("jdbc:mysql://hadoop102:3306/test")
                        .withUsername("root")
                        .withPassword("000000")
                        .build()
        );

        //旧的API
        ds.addSink(jdbcSink) ;

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
